package com.atguigu.edu.realtime.app.dws;

import com.atguigu.edu.realtime.app.func.KeywordUDTF;
import com.atguigu.edu.realtime.bean.KeywordBean;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MykafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        //基本环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //注册自定义的UTDF函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        //从kafka的topic_log中读取数据 创建动态表 指定WaterMark 提取事件事件
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_keyword_group";

        tableEnv.executeSql("CREATE TABLE page_log (\n" +
                "    common map<string,string>,\n" +
                "    page map<string,string>,\n" +
                "    ts BIGINT,\n" +
                "    row_time as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)) ,\n" +
                "    WATERMARK FOR row_time AS row_time - INTERVAL '3' SECOND\n" +
                ")" + MykafkaUtil.getKafkaDDL(topic,groupId));

        //过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select  page['item'] full_word, row_time\n" +
                "from page_log where page['item_type']='keyword' and page['item'] is not null");

        tableEnv.createTemporaryView("search_table",searchTable);
        //tableEnv.executeSql("select * from search_table").print();

        //使用自定义UTDF函数 对搜索内容分词 将分词结果和表中字段连接
        Table splitTable = tableEnv.sqlQuery("SELECT keyword,row_time FROM search_table,\n" +
                "LATERAL TABLE(ik_analyze(full_word)) t(keyword)");
        tableEnv.createTemporaryView("split_table",splitTable);

        //分组、开窗、聚合计算
        Table resTable = tableEnv.sqlQuery("select \n" +
                "    DATE_FORMAT(TUMBLE_START(row_time, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt,\n" +
                "    DATE_FORMAT(TUMBLE_END(row_time, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt,\n" +
                "    'SEARCH' source,\n" +
                "    keyword,\n" +
                "    count(*) keyword_count,\n" +
                "    UNIX_TIMESTAMP()*1000 ts\n" +
                "from split_table group by TUMBLE(row_time, INTERVAL '10' SECOND),keyword");
         tableEnv.createTemporaryView("res_table",resTable);
         tableEnv.executeSql("select * from res_table").print();

        //将动态表转换为流
        DataStream<KeywordBean> keywordDS = tableEnv.toAppendStream(resTable, KeywordBean.class);
        keywordDS.print(">>>>");

        //将流中的结果写到Clickhouse表中
        keywordDS.addSink(
                MyClickhouseUtil.getSinkFunction("insert into dws_traffic_source_keyword_page_view_window values(?,?,?,?,?,?)")
        );

        env.execute();
    }
}
